package drds.server.memory.unsafe.ringbuffer;

import drds.server.memory.unsafe.Platform;
import drds.server.memory.unsafe.memory.mm.MemoryConsumer;
import drds.server.memory.unsafe.ringbuffer.common.Cursored;
import drds.server.memory.unsafe.ringbuffer.common.event.EventFactory;
import drds.server.memory.unsafe.ringbuffer.common.event.EventSequencer;
import drds.server.memory.unsafe.ringbuffer.common.event.EventSink;
import drds.server.memory.unsafe.ringbuffer.common.event.EventTranslator;
import drds.server.memory.unsafe.ringbuffer.common.event.EventTranslatorOneArg;
import drds.server.memory.unsafe.ringbuffer.common.event.EventTranslatorThreeArg;
import drds.server.memory.unsafe.ringbuffer.common.event.EventTranslatorTwoArg;
import drds.server.memory.unsafe.ringbuffer.common.event.EventTranslatorVararg;
import drds.server.memory.unsafe.ringbuffer.exception.InsufficientCapacityException;
import drds.server.memory.unsafe.ringbuffer.producer.Sequencer;

/**
 * 环形buffer 待实现，
 */
public class RingBuffer<E extends MemoryConsumer> implements Cursored, EventSequencer<E>, EventSink<E> {
	// Buffer数组填充
	private static final int BUFFER_PAD;
	// Buffer数组起始基址
	private static final long REF_ARRAY_BASE;
	// 2^n=每个数组对象引用所占空间，这个n就是REF_ELEMENT_SHIFT
	private static final int REF_ELEMENT_SHIFT;

	static {
		final int scale = Platform.arrayIndexScale(Object[].class);
		// Object数组引用长度，32位为4字节，64位为8字节
		if (4 == scale) {
			REF_ELEMENT_SHIFT = 2;
		} else if (8 == scale) {
			REF_ELEMENT_SHIFT = 3;
		} else {
			throw new IllegalStateException("Unknown pointer size");
		}
		// 需要填充128字节，缓存行长度一般是128字节
		BUFFER_PAD = 128 / scale;
		REF_ARRAY_BASE = Platform.arrayBaseOffset(Object[].class) + (BUFFER_PAD << REF_ELEMENT_SHIFT);
	}

	private final long indexMask;
	private final Object[] entries;
	protected final int bufferSize;
	protected final Sequencer sequencer;

	public RingBuffer(EventFactory<E> eventFactory, Sequencer sequencer) {
		this.sequencer = sequencer;
		this.bufferSize = sequencer.getBufferSize();
		// 保证buffer大小不小于1
		if (bufferSize < 1) {
			throw new IllegalArgumentException("bufferSize must not be less than 1");
		}
		// 保证buffer大小为2的n次方
		if (Integer.bitCount(bufferSize) != 1) {
			throw new IllegalArgumentException("bufferSize must be a power of 2");
		}
		// m % 2^n <=> m & (2^n - 1)
		this.indexMask = bufferSize - 1;
		/**
		 * 结构：缓存行填充，避免频繁访问的任一entry与另一被修改的无关变量写入同一缓存行 -------------- * 数组头 * BASE
		 * * Padding * 128字节 * reference1 * SCALE * reference2 * SCALE *
		 * reference3 * SCALE .......... * Padding * 128字节 --------------
		 */
		this.entries = new Object[sequencer.getBufferSize() + 2 * BUFFER_PAD];
		// 利用eventFactory初始化RingBuffer的每个槽
		fill(eventFactory);
	}

	private void fill(EventFactory<E> eventFactory) {
		for (int i = 0; i < bufferSize; i++) {
			entries[BUFFER_PAD + i] = eventFactory.newInstance();
		}
	}

	/**
	 * 根据地址取出一个元素的引用
	 * 
	 * @param sequence
	 * @return
	 */
	private E elementAt(long sequence) {
		return (E) Platform.getObject(entries, REF_ARRAY_BASE + ((sequence & indexMask) << REF_ELEMENT_SHIFT));
	}

	@Override
	public long getCursor() {
		return sequencer.getCursor();
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.DataProvider#get(long)
	 */
	@Override
	public E get(long sequence) {
		return elementAt(sequence);
	}

	private void translateAndPublish(EventTranslator<E> translator, long sequence) {
		try {
			translator.translateTo(get(sequence), sequence);
		} finally {
			sequencer.publish(sequence);
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public void publishEvent(EventTranslator<E> translator) {
		final long sequence = sequencer.next();
		translateAndPublish(translator, sequence);
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#tryPublishEvent(EventTranslator)
	 */
	@Override
	public boolean tryPublishEvent(EventTranslator<E> translator) {
		try {
			final long sequence = sequencer.tryNext();
			translateAndPublish(translator, sequence);
			return true;
		} catch (InsufficientCapacityException e) {
			return false;
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	private <A> void translateAndPublish(EventTranslatorOneArg<E, A> translator, long sequence, A arg0) {
		try {
			translator.translateTo(get(sequence), sequence, arg0);
		} finally {
			sequencer.publish(sequence);
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A> void publishEvent(EventTranslatorOneArg<E, A> translator, A arg0) {
		final long sequence = sequencer.next();
		translateAndPublish(translator, sequence, arg0);
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A> boolean tryPublishEvent(EventTranslatorOneArg<E, A> translator, A arg0) {
		try {
			final long sequence = sequencer.tryNext();
			translateAndPublish(translator, sequence, arg0);
			return true;
		} catch (InsufficientCapacityException e) {
			return false;
		}
	}

	private <A, B> void translateAndPublish(EventTranslatorTwoArg<E, A, B> translator, long sequence, A arg0, B arg1) {
		try {
			translator.translateTo(get(sequence), sequence, arg0, arg1);
		} finally {
			sequencer.publish(sequence);
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A, B> void publishEvent(EventTranslatorTwoArg<E, A, B> translator, A arg0, B arg1) {
		final long sequence = sequencer.next();
		translateAndPublish(translator, sequence, arg0, arg1);
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A, B> boolean tryPublishEvent(EventTranslatorTwoArg<E, A, B> translator, A arg0, B arg1) {
		try {
			final long sequence = sequencer.tryNext();
			translateAndPublish(translator, sequence, arg0, arg1);
			return true;
		} catch (InsufficientCapacityException e) {
			return false;
		}
	}

	private <A, B, C> void translateAndPublish(EventTranslatorThreeArg<E, A, B, C> translator, long sequence, A arg0, B arg1, C arg2) {
		try {
			translator.translateTo(get(sequence), sequence, arg0, arg1, arg2);
		} finally {
			sequencer.publish(sequence);
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A, B, C> void publishEvent(EventTranslatorThreeArg<E, A, B, C> translator, A arg0, B arg1, C arg2) {
		final long sequence = sequencer.next();
		translateAndPublish(translator, sequence, arg0, arg1, arg2);
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A, B, C> boolean tryPublishEvent(EventTranslatorThreeArg<E, A, B, C> translator, A arg0, B arg1, C arg2) {
		try {
			final long sequence = sequencer.tryNext();
			translateAndPublish(translator, sequence, arg0, arg1, arg2);
			return true;
		} catch (InsufficientCapacityException e) {
			return false;
		}
	}

	private void translateAndPublish(EventTranslatorVararg<E> translator, long sequence, Object... args) {
		try {
			translator.translateTo(get(sequence), sequence, args);
		} finally {
			sequencer.publish(sequence);
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public void publishEvent(EventTranslatorVararg<E> translator, Object... args) {
		final long sequence = sequencer.next();
		translateAndPublish(translator, sequence, args);
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public boolean tryPublishEvent(EventTranslatorVararg<E> translator, Object... args) {
		try {
			final long sequence = sequencer.tryNext();
			translateAndPublish(translator, sequence, args);
			return true;
		} catch (InsufficientCapacityException e) {
			return false;
		}
	}

	private void checkBounds(final EventTranslator<E>[] translators, final int batchStartsAt, final int batchSize) {
		checkBatchSizing(batchStartsAt, batchSize);
		batchOverRuns(translators, batchStartsAt, batchSize);
	}

	private <A> void batchOverRuns(final A[] arg0, final int batchStartsAt, final int batchSize) {
		if (batchStartsAt + batchSize > arg0.length) {
			throw new IllegalArgumentException("A batchSize of: " + batchSize + " with batchStatsAt of: " + batchStartsAt + " will overrun the available number of arguments: " + (arg0.length - batchStartsAt));
		}
	}

	private void checkBatchSizing(int batchStartsAt, int batchSize) {
		if (batchStartsAt < 0 || batchSize < 0) {
			throw new IllegalArgumentException("Both batchStartsAt and batchSize must be positive but got: batchStartsAt " + batchStartsAt + " and batchSize " + batchSize);
		} else if (batchSize > bufferSize) {
			throw new IllegalArgumentException("The ring buffer cannot accommodate " + batchSize + " it only has space for " + bufferSize + " entities.");
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public void publishEvents(EventTranslator<E>[] translators) {
		publishEvents(translators, 0, translators.length);
	}

	private void translateAndPublishBatch(final EventTranslator<E>[] translators, int batchStartsAt, final int batchSize, final long finalSequence) {
		final long initialSequence = finalSequence - (batchSize - 1);
		try {
			long sequence = initialSequence;
			final int batchEndsAt = batchStartsAt + batchSize;
			for (int i = batchStartsAt; i < batchEndsAt; i++) {
				final EventTranslator<E> translator = translators[i];
				translator.translateTo(get(sequence), sequence++);
			}
		} finally {
			sequencer.publish(initialSequence, finalSequence);
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public void publishEvents(EventTranslator<E>[] translators, int batchStartsAt, int batchSize) {
		checkBounds(translators, batchStartsAt, batchSize);
		final long finalSequence = sequencer.next(batchSize);
		translateAndPublishBatch(translators, batchStartsAt, batchSize, finalSequence);
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public boolean tryPublishEvents(EventTranslator<E>[] translators) {
		return false;
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public boolean tryPublishEvents(EventTranslator<E>[] translators, int batchStartsAt, int batchSize) {
		checkBounds(translators, batchStartsAt, batchSize);
		try {
			final long finalSequence = sequencer.tryNext(batchSize);
			translateAndPublishBatch(translators, batchStartsAt, batchSize, finalSequence);
			return true;
		} catch (InsufficientCapacityException e) {
			return false;
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A> void publishEvents(EventTranslatorOneArg<E, A> translator, A[] arg0) {
		publishEvents(translator, 0, arg0.length, arg0);
	}

	private <A> void checkBounds(final A[] arg0, final int batchStartsAt, final int batchSize) {
		checkBatchSizing(batchStartsAt, batchSize);
		batchOverRuns(arg0, batchStartsAt, batchSize);
	}

	private <A> void translateAndPublishBatch(final EventTranslatorOneArg<E, A> translator, final A[] arg0, int batchStartsAt, final int batchSize, final long finalSequence) {
		final long initialSequence = finalSequence - (batchSize - 1);
		try {
			long sequence = initialSequence;
			final int batchEndsAt = batchStartsAt + batchSize;
			for (int i = batchStartsAt; i < batchEndsAt; i++) {
				translator.translateTo(get(sequence), sequence++, arg0[i]);
			}
		} finally {
			sequencer.publish(initialSequence, finalSequence);
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A> void publishEvents(EventTranslatorOneArg<E, A> translator, int batchStartsAt, int batchSize, A[] arg0) {
		checkBounds(arg0, batchStartsAt, batchSize);
		final long finalSequence = sequencer.next(batchSize);
		translateAndPublishBatch(translator, arg0, batchStartsAt, batchSize, finalSequence);
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A> boolean tryPublishEvents(EventTranslatorOneArg<E, A> translator, A[] arg0) {
		return tryPublishEvents(translator, 0, arg0.length, arg0);
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A> boolean tryPublishEvents(EventTranslatorOneArg<E, A> translator, int batchStartsAt, int batchSize, A[] arg0) {
		checkBounds(arg0, batchStartsAt, batchSize);
		try {
			final long finalSequence = sequencer.tryNext(batchSize);
			translateAndPublishBatch(translator, arg0, batchStartsAt, batchSize, finalSequence);
			return true;
		} catch (InsufficientCapacityException e) {
			return false;
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A, B> void publishEvents(EventTranslatorTwoArg<E, A, B> translator, A[] arg0, B[] arg1) {
		publishEvents(translator, 0, arg0.length, arg0, arg1);
	}

	private <A, B> void checkBounds(final A[] arg0, final B[] arg1, final int batchStartsAt, final int batchSize) {
		checkBatchSizing(batchStartsAt, batchSize);
		batchOverRuns(arg0, batchStartsAt, batchSize);
		batchOverRuns(arg1, batchStartsAt, batchSize);
	}

	private <A, B> void translateAndPublishBatch(final EventTranslatorTwoArg<E, A, B> translator, final A[] arg0, final B[] arg1, int batchStartsAt, int batchSize, final long finalSequence) {
		final long initialSequence = finalSequence - (batchSize - 1);
		try {
			long sequence = initialSequence;
			final int batchEndsAt = batchStartsAt + batchSize;
			for (int i = batchStartsAt; i < batchEndsAt; i++) {
				translator.translateTo(get(sequence), sequence++, arg0[i], arg1[i]);
			}
		} finally {
			sequencer.publish(initialSequence, finalSequence);
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A, B> void publishEvents(EventTranslatorTwoArg<E, A, B> translator, int batchStartsAt, int batchSize, A[] arg0, B[] arg1) {
		checkBounds(arg0, arg1, batchStartsAt, batchSize);
		final long finalSequence = sequencer.next(batchSize);
		translateAndPublishBatch(translator, arg0, arg1, batchStartsAt, batchSize, finalSequence);
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A, B> boolean tryPublishEvents(EventTranslatorTwoArg<E, A, B> translator, A[] arg0, B[] arg1) {
		return tryPublishEvents(translator, 0, arg0.length, arg0, arg1);
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A, B> boolean tryPublishEvents(EventTranslatorTwoArg<E, A, B> translator, int batchStartsAt, int batchSize, A[] arg0, B[] arg1) {
		checkBounds(arg0, arg1, batchStartsAt, batchSize);
		try {
			final long finalSequence = sequencer.tryNext(batchSize);
			translateAndPublishBatch(translator, arg0, arg1, batchStartsAt, batchSize, finalSequence);
			return true;
		} catch (InsufficientCapacityException e) {
			return false;
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A, B, C> void publishEvents(EventTranslatorThreeArg<E, A, B, C> translator, A[] arg0, B[] arg1, C[] arg2) {
		publishEvents(translator, 0, arg0.length, arg0, arg1, arg2);
	}

	private <A, B, C> void checkBounds(final A[] arg0, final B[] arg1, final C[] arg2, final int batchStartsAt, final int batchSize) {
		checkBatchSizing(batchStartsAt, batchSize);
		batchOverRuns(arg0, batchStartsAt, batchSize);
		batchOverRuns(arg1, batchStartsAt, batchSize);
		batchOverRuns(arg2, batchStartsAt, batchSize);
	}

	private <A, B, C> void translateAndPublishBatch(final EventTranslatorThreeArg<E, A, B, C> translator, final A[] arg0, final B[] arg1, final C[] arg2, int batchStartsAt, final int batchSize, final long finalSequence) {
		final long initialSequence = finalSequence - (batchSize - 1);
		try {
			long sequence = initialSequence;
			final int batchEndsAt = batchStartsAt + batchSize;
			for (int i = batchStartsAt; i < batchEndsAt; i++) {
				translator.translateTo(get(sequence), sequence++, arg0[i], arg1[i], arg2[i]);
			}
		} finally {
			sequencer.publish(initialSequence, finalSequence);
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A, B, C> void publishEvents(EventTranslatorThreeArg<E, A, B, C> translator, int batchStartsAt, int batchSize, A[] arg0, B[] arg1, C[] arg2) {
		checkBounds(arg0, arg1, arg2, batchStartsAt, batchSize);
		final long finalSequence = sequencer.next(batchSize);
		translateAndPublishBatch(translator, arg0, arg1, arg2, batchStartsAt, batchSize, finalSequence);
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A, B, C> boolean tryPublishEvents(EventTranslatorThreeArg<E, A, B, C> translator, A[] arg0, B[] arg1, C[] arg2) {
		return tryPublishEvents(translator, 0, arg0.length, arg0, arg1, arg2);
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public <A, B, C> boolean tryPublishEvents(EventTranslatorThreeArg<E, A, B, C> translator, int batchStartsAt, int batchSize, A[] arg0, B[] arg1, C[] arg2) {
		checkBounds(arg0, arg1, arg2, batchStartsAt, batchSize);
		try {
			final long finalSequence = sequencer.tryNext(batchSize);
			translateAndPublishBatch(translator, arg0, arg1, arg2, batchStartsAt, batchSize, finalSequence);
			return true;
		} catch (InsufficientCapacityException e) {
			return false;
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public void publishEvents(EventTranslatorVararg<E> translator, Object[]... args) {
		publishEvents(translator, 0, args.length, args);
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public void publishEvents(EventTranslatorVararg<E> translator, int batchStartsAt, int batchSize, Object[]... args) {
		checkBounds(batchStartsAt, batchSize, args);
		final long finalSequence = sequencer.next(batchSize);
		translateAndPublishBatch(translator, batchStartsAt, batchSize, finalSequence, args);
	}

	private void checkBounds(final int batchStartsAt, final int batchSize, final Object[][] args) {
		checkBatchSizing(batchStartsAt, batchSize);
		batchOverRuns(args, batchStartsAt, batchSize);
	}

	private void translateAndPublishBatch(final EventTranslatorVararg<E> translator, int batchStartsAt, final int batchSize, final long finalSequence, final Object[][] args) {
		final long initialSequence = finalSequence - (batchSize - 1);
		try {
			long sequence = initialSequence;
			final int batchEndsAt = batchStartsAt + batchSize;
			for (int i = batchStartsAt; i < batchEndsAt; i++) {
				translator.translateTo(get(sequence), sequence++, args[i]);
			}
		} finally {
			sequencer.publish(initialSequence, finalSequence);
		}
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public boolean tryPublishEvents(EventTranslatorVararg<E> translator, Object[]... args) {
		return tryPublishEvents(translator, 0, args.length, args);
	}

	/**
	 * @see drds.server.memory.unsafe.ringbuffer.common.event.EventSink#publishEvent(EventTranslator)
	 */
	@Override
	public boolean tryPublishEvents(EventTranslatorVararg<E> translator, int batchStartsAt, int batchSize, Object[]... args) {
		return false;
	}

	@Override
	public int getBufferSize() {
		return bufferSize;
	}

	@Override
	public boolean hasAvailableCapacity(int requiredCapacity) {
		return sequencer.hasAvailableCapacity(requiredCapacity);
	}

	@Override
	public long remainingCapacity() {
		return sequencer.remainingCapacity();
	}

	@Override
	public long next() {
		return sequencer.next();
	}

	@Override
	public long next(int n) {
		return sequencer.next(n);
	}

	@Override
	public long tryNext() throws InsufficientCapacityException {
		return sequencer.tryNext();
	}

	@Override
	public long tryNext(int n) throws InsufficientCapacityException {
		return sequencer.tryNext(n);
	}

	@Override
	public void publish(long sequence) {
		sequencer.publish(sequence);
	}

	@Override
	public void publish(long lo, long hi) {
		sequencer.publish(lo, hi);
	}
}
